#ifndef OTL_THREAD_QUEUE_H
#define OTL_THREAD_QUEUE_H

#include <iostream>
#include <string>
#include <vector>
#include <queue>
#include <deque>
#include <functional>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <chrono>

#if  defined(__linux__) || defined(__APPLE__)

#include <sys/time.h>

#endif

#include <pthread.h>
#include "otl_baseclass.h"
#include "otl_log.h"

namespace enrigin
{
    static int cpu_index = 0;

    template <typename T>
    class BlockingQueue : public NoCopyable
    {
    private:
        size_t size_impl() const
        {
            return m_type == 0 ? m_queue.size() : m_vec.size();
        }

        void wait_and_push_one(T&& data)
        {
            if (m_limit > 0 && this->size_impl() >= m_limit && !m_stop)
            {
# if USE_DEBUG
            OTL_LOGW(m_name.c_str(), "queue_size(%zu) > %d", this->size_impl(), m_limit);
# endif
                // flow control by dropping
                if (m_drop_fn != nullptr)
                {
                    this->drop_half_();
# if USE_DEBUG
                OTL_LOGW(m_name.c_str(), "queue_size after dropping, size: %zu", this->size_impl());
# endif
                }
                else
                {
                    // blocking
                    do
                    {
                        pthread_cond_wait(&m_push_condv, &m_qmtx);
                    }
                    while (m_limit > 0 && this->size_impl() >= m_limit && !m_stop);
                }
            }
            else if (this->size_impl() >= m_warning && !m_stop && this->size_impl() % 100 == 0)
            {
                OTL_LOGW(m_name.c_str(), "queue_size is %zu", this->size_impl());
            }

            if (m_type == 0)
            {
                m_queue.push(std::move(data));
            }
            else
            {
                m_vec.push_back(std::move(data));
            }
        }

    public:
        BlockingQueue(const std::string& name = "", int type = 0, int limit = 0, int warning = 32)
            : m_stop(false), m_limit(limit), m_drop_fn(nullptr), m_warning(warning)
        {
            m_name = name;
            m_type = type;
            pthread_mutex_init(&m_qmtx, NULL);
            pthread_cond_init(&m_push_condv, NULL);
            pthread_cond_init(&m_pop_condv, NULL);
        }

        ~BlockingQueue()
        {
            pthread_mutex_lock(&m_qmtx);
            OTL_LOGI(m_name.c_str(), "destroy, size: %zu", m_queue.size() + m_vec.size());
            m_vec.clear();
            std::queue<T> empty;
            m_queue.swap(empty);
            pthread_mutex_unlock(&m_qmtx);
        }

        void stop()
        {
            pthread_mutex_lock(&m_qmtx);
            m_stop = true;
            OTL_LOGI(m_name.c_str(), "stop blocking queue");
            pthread_cond_broadcast(&m_push_condv);
            pthread_cond_broadcast(&m_pop_condv);
            pthread_mutex_unlock(&m_qmtx);
        }

        int push(T& data)
        {
            pthread_mutex_lock(&m_qmtx);

            this->wait_and_push_one(std::move(data));
            int num = this->size_impl();
            pthread_cond_broadcast(&m_pop_condv);

            pthread_mutex_unlock(&m_qmtx);

            return num;
        }

        int push(std::vector<T>& datas)
        {
            int num;
            pthread_mutex_lock(&m_qmtx);

            for (auto& data : datas)
            {
                this->wait_and_push_one(std::move(data));
                if (m_stop) goto err;
                pthread_cond_signal(&m_pop_condv);
            }
            num = this->size_impl();

            pthread_mutex_unlock(&m_qmtx);
            return num;

        err:
            pthread_mutex_unlock(&m_qmtx);
            return 0;
        }

        int pop_front(std::vector<T>& objs, int min_num, int max_num, long wait_ms = 0, bool* p_is_timeout = nullptr)
        {
            bool is_timeout = false;

            struct timespec to;
            struct timeval now;
            gettimeofday(&now, NULL);
            double ms0 = now.tv_sec * 1000 + now.tv_usec / 1000.0;
            //std::cout << m_name << ",pop:" << now.tv_usec / 1000.0 << std::endl;
            if (wait_ms == 0)
            {
                to.tv_sec = now.tv_sec + 9999999;
                to.tv_nsec = now.tv_usec * 1000UL;
            }
            else
            {
                int nsec = now.tv_usec * 1000 + (wait_ms % 1000) * 1000000;
                to.tv_sec = now.tv_sec + nsec / 1000000000 + wait_ms / 1000;
                to.tv_nsec = nsec % 1000000000; //(now.tv_usec + wait_ms * 1000UL) * 1000UL;
            }
            pthread_mutex_lock(&m_qmtx);
            if (p_is_timeout) *p_is_timeout = false;
            while ((m_type ? m_vec.size() < min_num : m_queue.size() < min_num) && !m_stop)
            {
#ifdef BLOCKING_QUEUE_PERF
            m_timer.tic();
#endif
                // pthread_timestruc_t to;
                int err = pthread_cond_timedwait(&m_pop_condv, &m_qmtx, &to);
                if (err == ETIMEDOUT)
                {
                    is_timeout = true;
                    break;
                }
#ifdef BLOCKING_QUEUE_PERF
            m_timer.toc();
    if (m_timer.total_time_ > 1) {
      m_timer.summary();
    }
#endif
            }

            if (!is_timeout)
            {
                if (m_type == 0)
                {
                    int oc = 0;
                    while (oc < max_num && m_queue.size() > 0)
                    {
                        auto o = std::move(m_queue.front());
                        m_queue.pop();
                        objs.push_back(std::move(o));
                        oc++;
                    }
                }
                else
                {
                    int oc = 0;
                    while (oc < max_num && !m_vec.empty())
                    {
                        auto o = std::move(m_vec.front());
                        m_vec.pop_front();
                        objs.push_back(std::move(o));
                        oc++;
                    }
                }
                pthread_cond_broadcast(&m_push_condv);
            }

            pthread_mutex_unlock(&m_qmtx);

            if (m_stop)
            {
                return 0;
            }

            if (is_timeout)
            {
                if (p_is_timeout) *p_is_timeout = true;
                return -1;
            }

            return 0;
        }

        size_t size()
        {
            size_t queue_size;
            pthread_mutex_lock(&m_qmtx);
            queue_size = this->size_impl();
            pthread_mutex_unlock(&m_qmtx);
            return queue_size;
        }

        int set_drop_fn(std::function<void(T& obj)> fn)
        {
            m_drop_fn = fn;
            return m_limit;
        }

        void drop_half_()
        {
            if (m_type == 0)
            {
                std::queue<T> temp;
                size_t num = m_queue.size();
                for (size_t i = 0; i < num; i++)
                {
                    auto elem = std::move(m_queue.front());
                    m_queue.pop();
                    if ((i % 2) == 0) {
                        temp.push(std::move(elem));
                    } else if (m_drop_fn) {
                        m_drop_fn(elem);
                    }
                }
                m_queue.swap(temp);
            }
            else
            {
                std::deque<T> temp;
                size_t num = m_vec.size();
                for (size_t i = 0; i < num; i++)
                {
                    auto elem = std::move(m_vec.front());
                    m_vec.pop_front();
                    if ((i % 2) == 0) {
                        temp.push_back(std::move(elem));
                    } else if (m_drop_fn) {
                        m_drop_fn(elem);
                    }
                }
                m_vec.swap(temp);
            }
        }

        void drop(int num = 0)
        {
            int queue_size;
            pthread_mutex_lock(&m_qmtx);
            if (num == 0)
            {
                num = this->size_impl();
            }
            if (this->size_impl() < num) {
                pthread_mutex_unlock(&m_qmtx);
                return;
            }
            if (m_type == 0)
            {
                queue_size = m_queue.size();
                if (num > queue_size)
                    num = queue_size;
                for (int i = 0; i < num; i++)
                {
                    m_queue.pop();
                }
            }
            else
            {
                queue_size = m_vec.size();
                if (num > queue_size)
                    num = queue_size;
                for (int i = 0; i < num; ++i) {
                    m_vec.pop_front();
                }
            }
            pthread_cond_broadcast(&m_push_condv);
            pthread_mutex_unlock(&m_qmtx);
        }

        const std::string& name() { return m_name; }

    private:
        bool m_stop;
        std::string m_name;
        std::deque<T> m_vec; // use deque for efficient pop_front
        std::queue<T> m_queue;
        pthread_mutex_t m_qmtx;
        pthread_cond_t m_pop_condv;
        pthread_cond_t m_push_condv;
        int m_type, m_limit; //0:queue,1:vector
        int m_warning;
        std::function<void(T& obj)> m_drop_fn;
    };

    // Lightweight blocking queue for generic use-cases (simple push/pop with optional timeout)
    // Note: distinct from enrigin::BlockingQueue above. This variant matches simple semantics used by stream pusher.
    namespace internal {
        template<typename T>
        class BlockingQueue {
        private:
            std::queue<T> m_queue;
            mutable std::mutex m_mutex;
            std::condition_variable m_condition;
            bool m_shutdown{false};

        public:
            void push(T item) {
                std::lock_guard<std::mutex> lock(m_mutex);
                if (!m_shutdown) {
                    m_queue.push(std::move(item));
                    m_condition.notify_one();
                }
            }

            bool pop(T& item, int timeoutMs = -1) {
                std::unique_lock<std::mutex> lock(m_mutex);
                if (timeoutMs < 0) {
                    m_condition.wait(lock, [this] { return !m_queue.empty() || m_shutdown; });
                } else {
                    if (!m_condition.wait_for(lock, std::chrono::milliseconds(timeoutMs),
                                              [this] { return !m_queue.empty() || m_shutdown; })) {
                        return false; // timeout
                    }
                }
                if (m_shutdown && m_queue.empty()) return false;
                if (!m_queue.empty()) {
                    item = m_queue.front();
                    m_queue.pop();
                    return true;
                }
                return false;
            }

            size_t size() const {
                std::lock_guard<std::mutex> lock(m_mutex);
                return m_queue.size();
            }

            bool empty() const {
                std::lock_guard<std::mutex> lock(m_mutex);
                return m_queue.empty();
            }

            void shutdown() {
                std::lock_guard<std::mutex> lock(m_mutex);
                m_shutdown = true;
                m_condition.notify_all();
            }

            void reset() {
                std::lock_guard<std::mutex> lock(m_mutex);
                m_shutdown = false;
                while (!m_queue.empty()) m_queue.pop();
            }
        };
    }

    template <typename T>
    class WorkerPool : public NoCopyable
    {
        BlockingQueue<T>* m_work_que;
        int m_thread_num;
        bool m_thread_running{true};
        using OnWorkItemsCallback = std::function<void(std::vector<T>& item)>;
        OnWorkItemsCallback m_work_item_func;
      
        std::vector<std::thread*> m_threads;
        int m_max_pop_num;
        int m_min_pop_num;
        // Pre-start tasks: can be posted before startWork(); threads will execute them first
        std::deque<std::function<void()>> m_prestart_tasks;
        std::mutex m_prestart_mtx;

    public:
        WorkerPool() : m_work_que(nullptr), m_thread_num(0), m_work_item_func(nullptr), m_max_pop_num(1),
                       m_min_pop_num(1)
        {
        }

        virtual ~WorkerPool()
        {
            for (auto pth : m_threads)
            {
                pth->join();
                delete pth;
            }
        }

        int init(BlockingQueue<T>* que, int thread_num, int min_pop_num, int max_pop_num)
        {
            m_work_que = que;
            m_thread_num = thread_num;
            m_min_pop_num = min_pop_num;
            m_max_pop_num = max_pop_num;
            return 0;
        }

        int startWork(OnWorkItemsCallback loop_func)
        {
            m_work_item_func = loop_func;

            for (int i = 0; i < m_thread_num; ++i)
            {
                auto pth = new std::thread([this]
                {
                    // Drain and execute any pre-submitted startup tasks before normal loop
                    while (true)
                    {
                        std::function<void()> task;
                        {
                            std::lock_guard<std::mutex> lk(m_prestart_mtx);
                            if (m_prestart_tasks.empty()) break;
                            task = std::move(m_prestart_tasks.front());
                            m_prestart_tasks.pop_front();
                        }
                        if (task) task();
                    }

                    while (m_thread_running)
                    {
                        std::vector<T> items;

                        //if (m_work_que->size() < 4) { bm::usleep(10); continue; }
                        if (m_work_que->pop_front(items, m_min_pop_num, m_max_pop_num) != 0)
                        {
                            break;
                        }
                        if (items.empty())
                            break;
                        m_work_item_func(items);
                    }
                });
                //setCPU(*pth);
                m_threads.push_back(pth);
            }
            return 0;
        }

        void setCPU(std::thread& th)
        {
            static auto cpu_count = std::thread::hardware_concurrency();
#ifndef __APPLE__
            cpu_set_t cpuset;
            CPU_ZERO(&cpuset);
            CPU_SET(cpu_index++ % cpu_count, &cpuset);
            int ret = pthread_setaffinity_np(th.native_handle(),
                                             sizeof(cpu_set_t),
                                             &cpuset);
            if (ret != 0)
            {
                std::cerr << "[ERROR] caling pthread_setaffinity_np failed" << std::endl;
                exit(-1);
            }
            else
            {
                std::cout << "[SUCCESS] caling pthread_setaffinity_np success, " << cpu_index << std::endl;
            }
#endif
        }

        int stopWork()
        {
            m_work_que->stop();
            for (int i = 0; i < m_thread_num; i++)
            {
                m_threads[i]->join();
                delete m_threads[i];
                m_threads[i] = nullptr;
            }
            return 0;
        }

        int flush()
        {
            m_work_que->stop();
            return 0;
        }

        // New: enqueue a startup task to be executed before normal work when threads start
        int enqueue_startup_task(std::function<void()> fn)
        {
            std::lock_guard<std::mutex> lk(m_prestart_mtx);
            m_prestart_tasks.emplace_back(std::move(fn));
            return static_cast<int>(m_prestart_tasks.size());
        }
    };
} // namespace enrigin

#if !defined(ENRIGIN_NO_COMPAT_ALIAS)
#ifndef ENRIGIN_ALIAS_DEFINED
#define ENRIGIN_ALIAS_DEFINED
namespace otl = enrigin;
#endif
#endif

#endif // OTL_THREAD_QUEUE_H
